import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.Statement;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import com.alibaba.fastjson.JSON;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;

public class bolt2_2 extends BaseRichBolt {
    private static final long serialVersionUID = 881249197481637895L;
    private OutputCollector collector;
    private Map<String, AtomicInteger> counterMap;
    private int count;
    private double sum_price;
    private Map<String, Integer> guest = new HashMap<String, Integer>();
    private Map<String, Integer> zone = new HashMap<String, Integer>();
    private Map<String, Integer> pay_sum = new HashMap<String, Integer>();


    public void prepare(Map stormConf, TopologyContext context,
                        OutputCollector collector) {
        this.collector = collector;
        this.counterMap = new HashMap<String, AtomicInteger>();
    }

    public void execute(Tuple input) {
        try{
            //声名变量用于接收最终的结果数据
            String list1 = null;
            String list2 = null;
            String list1_1 = null;
            String list2_1 = null;
            //接收上一个bolt中传过来的数据
            String total_price_1 = input.getStringByField("total_price");
            String guest_name=input.getStringByField("guest_name");
            String receiveplace=input.getStringByField("receiveplace");
            String pay=input.getStringByField("pay");
            int all=input.getIntegerByField("all");
            System.out.println(pay);
            float total_price=Float.parseFloat(total_price_1);;
            //计算总销售额
            sum_price+=total_price;
            //当前总下单客户数
            int num=0;
            if (guest.containsKey(guest_name)) {
                num = guest.get(guest_name);
            }
            num+=1;
            guest.put(guest_name,num);

            //求付款类型
            int num_3=0;
            if (pay_sum.containsKey(pay)) {
                num_3 = pay_sum.get(pay);
            }
            num_3+=1;
            pay_sum.put(pay,num_3);
            String regEx = "[\n`\"\"/”“\\\\\\\\]";
            //统计付款类型
            try {
                //hash map按key进行从大到小的排序
                List<String> mobileList = pay_sum.entrySet().stream()
                        .sorted((Map.Entry<String, Integer> e1, Map.Entry<String, Integer> e2) -> e2.getValue() - e1.getValue())
                        .map(entry -> entry.getKey()).collect(Collectors.toList())
                        .subList(0, 3);
                List<Integer> mobileList_1 = pay_sum.entrySet().stream()
                        .sorted((Map.Entry<String, Integer> e1, Map.Entry<String, Integer> e2) -> e2.getValue() - e1.getValue())
                        .map(entry -> entry.getValue()).collect(Collectors.toList())
                        .subList(0, 3);
                list2_1=JSON.toJSONString(mobileList_1);
                list2=JSON.toJSONString(mobileList);
                list2 = list2.replaceAll(regEx, "");
                //System.out.println(list2_1);
                //System.out.println(list2);
            }catch (Exception e){}
            //统计各省份下单客户累计数量
            String p=addressResolution(receiveplace);
            int num_2=0;
            if (zone.containsKey(p)) {
                num_2 = zone.get(p);
            }
            num_2+=1;
            zone.put(p,num_2);
            //求出前10的省份销量
            try {
                //hash map按key进行从大到小的排序
                List<String> mobileList = zone.entrySet().stream()
                        .sorted((Map.Entry<String, Integer> e1, Map.Entry<String, Integer> e2) -> e2.getValue() - e1.getValue())
                        .map(entry -> entry.getKey()).collect(Collectors.toList())
                        .subList(0, 10);
                List<Integer> mobileList_1 = zone.entrySet().stream()
                        .sorted((Map.Entry<String, Integer> e1, Map.Entry<String, Integer> e2) -> e2.getValue() - e1.getValue())
                        .map(entry -> entry.getValue()).collect(Collectors.toList())
                        .subList(0, 10);
                list1_1=JSON.toJSONString(mobileList_1);
                list1=JSON.toJSONString(mobileList);
                //System.out.println(list1_1);
                //System.out.println(list1);
                list1 = list1.replaceAll(regEx, "");
                //将结果写入数据库中
                //以映射方式加载Driver,抛出异常
                Class.forName("com.mysql.jdbc.Driver");
                //建立连接
                Connection conn = DriverManager.getConnection("jdbc:mysql://slave1:3306/storm1?useUnicode=true&characterEncoding=gbk", "root", "root");
                //执行sql语句
                String sql="insert into order_detail (total,sum,provice,top_pro,top_pro_sum,pay,pay_sum) values ("+sum_price+","+all+",\""+zone+"\""+",\""+list1+"\""+",\""+list1_1+"\""+",\""+list2+"\""+",\""+list2_1+"\");";
                //通过创建的连接对象创建执行器,用来执行sql语句
                Statement statement = conn.createStatement();
                //用于调试输出sql语句
                //System.out.println(sql);
                statement.executeUpdate(sql);
                System.out.println(sql);
                conn.close();
                statement.close();
                //成功操作后返回ack信号
                collector.ack(input);
            }catch (Exception e){}
        }catch (Exception e){
            //操作失败后返回fail信号
            collector.fail(input);
        }
    }

    public void cleanup() {
        Iterator<Entry<String, AtomicInteger>> iter = this.counterMap
                .entrySet().iterator();
        while (iter.hasNext()) {
            Entry<String, AtomicInteger> entry = iter.next();
        }

    }

    //用于从地址数据中提取省份信息
    public static String addressResolution(String address){
        String province=null;
        String []zones = {"北京","天津","上海","重庆","河北","山西","辽宁","吉林","黑龙江","江苏",
                "浙江","安徽","福建","江西","山东","河南","湖北","湖南","广东","海南",
                "四川","贵州","云南","陕西","甘肃","青海","台湾","内蒙古","广西","西藏","宁夏","新疆","香港","澳门"};
        for(int i=0;i<zones.length;i++){
            if(address.contains(zones[i])){
                province=zones[i];
                break;
            }
        }
        return province;
    }
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
    }
}